package cn.zzp.pet.mq;

import cn.zzp.order.domain.OrderPetAcquisition;
import cn.zzp.order.enums.OrderEnum;
import cn.zzp.order.mapper.OrderPetAcquisitionMapper;
import cn.zzp.order.service.OrderPetAcquisitionService;
import cn.zzp.pet.domain.Pet;
import cn.zzp.pet.domain.SearchMasterMsg;
import cn.zzp.pet.enums.SearchMasterMsgEnum;
import cn.zzp.pet.mapper.PetMapper;
import cn.zzp.pet.mapper.SearchMasterMsgMapper;
import cn.zzp.pet.service.SearchMasterMsgService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.xml.ws.soap.Addressing;
import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @ClassName Consumer
 * @Description 延时消息自动启动接收处理
 * @Author 祝志鹏
 * @Date 2022/8/19 0:51
 */
@Component
public class Consumer implements CommandLineRunner {

    @Autowired
    private OrderPetAcquisitionMapper orderPetAcquisitionMapper;

    @Autowired
    private SearchMasterMsgMapper searchMasterMsgMapper;

    @Autowired
    private PetMapper petMapper;

    @Autowired
    private OrderPetAcquisitionService orderPetAcquisitionService;

    @Value("${rocketmq.producer.group}")
    private String group;

    @Value("${rocketmq.name-server}")
    private String nameServerAddr;

    @Value("${rocketmq.producer.topic}")
    private String topic;

    @Value("${rocketmq.producer.tag}")
    private String tag;

    // 定义接收消息
    @Transactional
    public void consumerListener(){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
        consumer.setNamesrvAddr(nameServerAddr);

        // 订阅主题
        try {
            consumer.subscribe(topic, tag);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    for (MessageExt messageExt : list) {
                        String orderSn = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                        // 根据订单编号查询对象
                        OrderPetAcquisition order = orderPetAcquisitionMapper.queryBySn(orderSn);
                        // 判断是否处理，未处理将订单放回寻主池
                        if (order.getState() == 0){ // 未处理，默认取消订单操作
                            order.setState(OrderEnum.CANCEL.getState());
                            orderPetAcquisitionMapper.update(order);

                            // 修改消息发布
                            SearchMasterMsg smm = searchMasterMsgMapper.queryById(order.getSearch_master_msg_id());
                            smm.setShop_id(null);
                            smm.setState(SearchMasterMsgEnum.REJECT.getState());
                            smm.setNote(SearchMasterMsgEnum.REJECT.getDes());
                            searchMasterMsgMapper.update(smm);
                            // 删除宠物
                            Pet pet = petMapper.queryById(order.getPet_id());
                            petMapper.delete(pet.getId());
                        }
                    }

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void run(String... args) throws Exception {
        consumerListener();
    }
}
